package com.gongyibao.gbase.mq.topic2;

/**
 * @Author: 张启峰
 * @Date: 2023/4/17 14:27
 * @Description: gyb-
 */

import com.cvicse.inforsuitemq.InforsuiteMQConnection;
import com.cvicse.inforsuitemq.InforsuiteMQConnectionFactory;

import javax.jms.*;

public class Topic_Subscribe {
    private String topicname = "bb";//主题名
    private boolean transacted = false;//是否是事务消息
    private int ackMode = Session.AUTO_ACKNOWLEDGE;//消息确认机制，AUTO_ACKNOWLEDGE为自动确认,CLIENT_ACKNOWLEDGE为手动确认
    private String url = "failover:(tcp://192.168.6.55:8688)";
    private long sleepTime = 1000;//消息接收间隔，便于测试用
    private String ClientID = "11";

    private Connection connection = null;
    private Session session;
    private Destination destination;
    private MessageConsumer consumer;

    public static void main(String[] args) {
        Topic_Subscribe subscribe = new Topic_Subscribe();
        subscribe.recvMessage();
    }

    // 接收消息
    public void recvMessage() {
        try {
            System.out.println("Starting example now...");
            InforsuiteMQConnectionFactory connectionFactory = new InforsuiteMQConnectionFactory(
                    InforsuiteMQConnection.DEFAULT_USER,
                    InforsuiteMQConnection.DEFAULT_PASSWORD, url);
            connection = connectionFactory.createConnection();
            connection.setClientID(ClientID);
            connection.start();
            session = connection.createSession(transacted, ackMode);
            destination = session.createTopic(topicname);
            consumer = session.createDurableSubscriber((Topic) destination, ClientID); //持久订阅
            while (true) {
                Message message = consumer.receive();
                onMessage(message);
            }
        } catch (Exception e) {
            System.out.println("Caught:" + e);
            e.printStackTrace();
        }
    }

    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage txtMsg = (TextMessage) message;
                String msg = txtMsg.getText();
                System.out.println("Received:" + msg);
            } else {
                System.out.println("Received:" + message);
            }
            if (transacted) {
                session.commit();
            } else if (ackMode == Session.CLIENT_ACKNOWLEDGE) {
                message.acknowledge();
            }
        } catch (JMSException e) {
            System.out.println("Caught:" + e);
            e.printStackTrace();
        } finally {
            if (sleepTime > 0) {
                try {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e) {
                }
            }
        }
    }
}
